package com.ssm.juc.thread;

import org.jetbrains.annotations.NotNull;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

public class BatchTaskManager {

    private static BatchTaskManager instance;

    public static BatchTaskManager getInstance() {
        return instance;
    }

    private BlockingDeque<TaskParam<String>> blockingDeque;
    private ScheduledThreadPoolExecutor scheduledExecutor;

    public BatchTaskManager() {
        // DCL
        if (instance == null) {
            synchronized (BatchTaskManager.class) {
                if (instance == null) {
                    instance = new BatchTaskManager();
                    init();
                }
            }
        }
    }

    private void init() {
        instance.blockingDeque = new LinkedBlockingDeque<>(20000);
        instance.scheduledExecutor = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
            int i = 0;

            @Override
            public Thread newThread(@NotNull Runnable r) {
                if (i >= Integer.MAX_VALUE) {
                    i = 0;
                }
                i++;
                return new Thread(r, "Scheduled-Task-" + i);
            }
        }, new ThreadPoolExecutor.CallerRunsPolicy());

        instance.scheduledExecutor.scheduleAtFixedRate(this::batchHandler, 0, 20, TimeUnit.SECONDS);
    }

    private void batchHandler() {
        try {
            int maxLen = 5000;
            List<TaskParam<String>> taskParamList = new ArrayList<>(maxLen);
            for (int j = 0; j < maxLen; j++) {
                TaskParam<String> taskParam = blockingDeque.poll();
                if (taskParam == null) {
                    break;
                }
                taskParamList.add(taskParam);
            }
            if (taskParamList.size() == 0) {
                return;
            }

            Map<String, List<TaskParam<String>>> taskParamMap =
                    taskParamList.stream()
                            .collect(Collectors.groupingBy(TaskParam::getKey));

            // 获取数据并且打印
            taskParamMap.forEach((key, list) -> {
                String title = "title : " + key;
                System.out.println(title);
                String context = "context : " + list.stream().map(String::valueOf).collect(Collectors.joining(","));
                System.out.println(context);
            });

            //map {"a", ["1", "2", "3"], "b", ["4", "5", "6"]}
            //list ["a#1,2,3, "b#4,5,6"]
            /*
            List<String> shortList = taskParamList.stream()
                    .collect(Collectors.groupingBy(TaskParam::getKey))
                    .entrySet().stream().map(item -> {
                        List<TaskParam<String>> list = item.getValue();
                        String values = list.stream().map(TaskParam::getData).map(String::valueOf).collect(Collectors.joining(","));
                        return item.getKey() + ":" + values;
                    }).collect(Collectors.toList());
            */

        } catch (Throwable t) {
            t.printStackTrace();
            System.err.println("err" + t);
        }
    }

    public void submitTask(TaskParam<String> taskParam) {
        try {
            this.blockingDeque.put(taskParam);
        } catch (InterruptedException e) {
            System.out.println("submitTask error" + e);
        }
    }

    public static void main(String[] args) {
        getInstance().submitTask(new TaskParam<>("a", ""));
    }

    static class TaskParam<T> {
        private String key;
        private T data;

        public TaskParam() {
        }

        public TaskParam(String key, T data) {
            this.key = key;
            this.data = data;
        }

        public String getKey() {
            return key;
        }

        public void setKey(String key) {
            this.key = key;
        }

        public T getData() {
            return data;
        }

        public void setData(T data) {
            this.data = data;
        }
    }
}
